package ex.datastream.functions.richFunction;

import ex.vo.SensorReading;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

public class MyProcessWindowFunction extends ProcessWindowFunction<SensorReading, Tuple2<Long,SensorReading>, String, TimeWindow> {


    @Override
    public void process(String s, Context context, Iterable<SensorReading> iterable, Collector<Tuple2<Long, SensorReading>> out) throws Exception {
        SensorReading min = iterable.iterator().next();
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String str=String.format("start:%s;end:%s;key:%s", dateFormat.format(context.window().getStart()),
                dateFormat.format(context.window().getEnd()),s);
        System.out.println(str);
        out.collect(new Tuple2<Long,SensorReading>(context.window().getEnd(),min));
    }
}
